package com.eascs.spider.yuqing.worker.manager;

import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.http.impl.client.CloseableHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.eascs.spider.yuqing.common.proj.Platform;
import com.eascs.spider.yuqing.common.proj.Task;
import com.eascs.spider.yuqing.common.util.HttpUtil;
import com.eascs.spider.yuqing.common.util.PropertiesUtil;
import com.eascs.spider.yuqing.worker.spider.GlobalSpider;
import com.eascs.spider.yuqing.worker.spider.IfengNewsSpider;
import com.eascs.spider.yuqing.worker.spider.NcInfoSpider;
import com.eascs.spider.yuqing.worker.spider.NcPriceSpider;
import com.eascs.spider.yuqing.worker.spider.NcSupplySpider;
import com.eascs.spider.yuqing.worker.spider.SinaNewsSpider;
import com.eascs.spider.yuqing.worker.spider.SinaWeiboSpider;
import com.eascs.spider.yuqing.worker.spider.ZolPhoneSpider;

/**
 * 工作节点任务管理器
 * @author hzp
 *
 */
public class TaskManager {

	private static Logger logger = LoggerFactory.getLogger(TaskManager.class);
	
	private Queue<Task> receiveQueue = new ConcurrentLinkedQueue<>();//用于接收从调度中心发送过来的任务
	
	private Queue<Task> receivePriorityQueue = new ConcurrentLinkedQueue<>();//较高优先级用于接收从调度中心发送过来的任务
	
	private Queue<Task> sendQueue = new ConcurrentLinkedQueue<>();//用于存储执行任务中新产生的url，及发送至调度中心的url
	
	private ExecutorService executors;//爬虫任务执行线程池
	
	private static TaskManager instance;
	
	private String schduleRecieveUrl = PropertiesUtil.getConfig("schdule.receive.path");//调度中心任务接收路径
	
	private CloseableHttpClient client = HttpUtil.initHttpClient();
	
	private static Set<String> duplicate = new HashSet<>();//用于进程内部第一次去重
	
	public synchronized static TaskManager getInstance(){
		return instance == null?instance = new TaskManager():instance;
	}
	
	private TaskManager(){
		init();
	}
	
	private void init(){
		//1,初始化爬虫任务执行线程池
		executors = Executors.newFixedThreadPool(30);
		ExecutorService executors_supply = Executors.newFixedThreadPool(10);//农商网供应信息店铺任务专用
		//2,初始化调度线程池,用于监控
		ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
		//2.1,工作节点状态监控线程
		executorService.scheduleAtFixedRate(new Runnable() {
			
			@Override
			public void run() {
				logger.info("工作节点监控,receiveQueue.size:{},sendQueue.size:{}" , receiveQueue.size() , sendQueue.size());
			}
		}, 5, 120, TimeUnit.SECONDS);
		
		//3.1,工作节点任务分发线程,负责轮询接收队列receiveQueue中任务
		executorService.scheduleAtFixedRate(new Runnable() {
			
			@Override
			public void run() {
				Task task = fetchTaskFromReveiveQueue();
				if(task != null){
					String url = task.getUrl();
					Platform platform = Platform.getPlatform(url);
					if(platform == null){
						logger.warn("没匹配到的url:{}" ,url);
						return;
					}
					Runnable spider = null;
					switch(platform.getPlatform()){
					case "sina_news":
						spider = new SinaNewsSpider(url);break;
					case "ifeng_news":
						spider = new IfengNewsSpider(url);break;
					case "sina_weibo":
						spider = new SinaWeiboSpider(url);break;
					case "nc_price":
						spider = new NcPriceSpider(task);break;
					case "nc_supply":
						spider = new NcSupplySpider(task);break;
					case "nc_info":
						spider = new NcInfoSpider(task);break;
					case "zol_phone":
						spider = new ZolPhoneSpider(task);break;
					default :
						spider = new GlobalSpider(url);
					}
					if(task.getUrl().contains("http://nc.mofcom.gov.cn/supply")){
						executors_supply.execute(spider);
					}else{
						executors.execute(spider);
					}
				}
			}
		}, 100, 5, TimeUnit.MILLISECONDS);
		
		//3.2 工作节点新增任务分发线程
		executorService.scheduleAtFixedRate(new Runnable() {
			
			@Override
			public void run() {
				if(!sendQueue.isEmpty()){
					Task task = fetchTaskFromSendQueue();
					HttpUtil.executePost(client, schduleRecieveUrl, task);
				}
			}
		}, 100, 50, TimeUnit.MILLISECONDS);
	}
	
	public void addReceiveQueue(Task task){
		//新农商网供应信息中店铺爬取任务优先级较高,下次添加其他优先任务时做个优先名单
		if(task.getUrl().contains("http://nc.mofcom.gov.cn/supply"))
			receivePriorityQueue.add(task);
		else 
			receiveQueue.add(task);
	}
	
	
	public void addSendQueue(Task task){
		if(task.getUrl().contains("http://nc.mofcom.gov.cn/supply")){
			if(duplicate.contains(task.getUrl()))
				return;
			else 
				duplicate.add(task.getUrl());
		}
		sendQueue.add(task);
	}
	
	public Task fetchTaskFromReveiveQueue(){
		return receivePriorityQueue.isEmpty()?receiveQueue.poll():receivePriorityQueue.poll();
	}
	
	public Task fetchTaskFromSendQueue(){
		return sendQueue.poll();
	}
	
}
